package flinkcep

import java.time.Duration
import java.util

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object DemoTest {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 获取登录事件流，并提取时间戳、生成水位线

    val stream = env
      .fromElements(
        LoginEvent("user_1", "192.168.0.1", "fail", 2000L),
        LoginEvent("user_1", "192.168.0.2", "fail", 3000L),
        LoginEvent("user_2", "192.168.1.29", "fail", 4000L),
        LoginEvent("user_1", "171.56.23.10", "fail", 5000L),
        LoginEvent("user_2", "192.168.1.29", "fail", 7000L),
        LoginEvent("user_2", "192.168.1.29", "fail", 8000L),
        LoginEvent("user_2", "192.168.1.29", "success", 6000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forBoundedOutOfOrderness(Duration.ofSeconds(3))
        .withTimestampAssigner(new SerializableTimestampAssigner[LoginEvent] {
          override def extractTimestamp(element: LoginEvent, recordTimestamp: Long): Long = element.ts
        }))
      .keyBy(_.userId)


    // 1. 定义 Pattern，连续的三个登录失败事件
//    val pattern = Pattern
//      // 以第一个登录失败事件开始
//      .begin[LoginEvent]("first")
//      .where(_.eventType.equals("fail"))
//      // 接着是第二个登录失败事件
//      .next("second")
//      .where(_.eventType.equals("fail"))
//      // 接着是第三个登录失败事件
//      .next("third")
//      .where(_.eventType.equals("fail"))

    val pattern=Pattern.begin[LoginEvent]("first")
      .where(_.eventType.equals("fail"))
      .times(3)
      .consecutive()
      .within(Time.seconds(10))

    // 2. 将 Pattern 应用到流上，检测匹配的复杂事件，得到一个
    val patternStream = CEP.pattern(stream, pattern)

    // 3. 将匹配到的复杂事件选择出来，然后包装成字符串报警信息输出
    patternStream.select(new PatternSelectFunction[LoginEvent, String] {
      override def select(map: util.Map[String, util.List[LoginEvent]]): String = {
        //val first = map.get("first").get(0)
        //val second = map.get("second").get(0)
        //val third = map.get("third").get(0)
        //first.userId + "连续三次登录失败！！！" + Array(first.ts, second.ts, third.ts).mkString(",")
        val first = map.get("first").get(0)
        val first_2 = map.get("first").get(1)
        val first_3 = map.get("first").get(2)
        first.userId+"连续登录三次失败！！！" + Array(first.ts,first_2.ts,first_3.ts).mkString(",")
      }
    }).print("warn! ")

    env.execute()
  }

}

case class LoginEvent(userId: String, ip: String, eventType: String, ts: Long)
